package org.apache.flink.runtime.resourceprovider;

import static org.apache.flink.util.Preconditions.checkArgument;
import static org.apache.flink.util.Preconditions.checkNotNull;

import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.net.InetSocketAddress;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeoutException;
import java.util.function.BiConsumer;
import java.util.stream.Collectors;

import javax.annotation.Nonnull;
import javax.annotation.Nullable;

import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.CommandListener;
import org.apache.flink.api.common.ConsumerID;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.ClusterOptions;
import org.apache.flink.management.jmx.JMXService;
import org.apache.flink.runtime.accumulators.AccumulatorSnapshot;
import org.apache.flink.runtime.allocatedeploy.execution.CommandDescriptor;
import org.apache.flink.runtime.allocatedeploy.execution.ExecutionID;
import org.apache.flink.runtime.blob.ConsumerPermanentBlobService;
import org.apache.flink.runtime.blob.TaskExecutorBlobService;
import org.apache.flink.runtime.blob.TransientBlobKey;
import org.apache.flink.runtime.blob.TransientBlobService;
import org.apache.flink.runtime.clusterframework.types.AllocationID;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
import org.apache.flink.runtime.clusterframework.types.SlotID;
import org.apache.flink.runtime.entrypoint.ClusterInformation;
import org.apache.flink.runtime.externalresource.ExternalResourceInfoProvider;
import org.apache.flink.runtime.filecache.FileCache;
import org.apache.flink.runtime.heartbeat.HeartbeatListener;
import org.apache.flink.runtime.heartbeat.HeartbeatManager;
import org.apache.flink.runtime.heartbeat.HeartbeatReceiver;
import org.apache.flink.runtime.heartbeat.HeartbeatServices;
import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
import org.apache.flink.runtime.instance.HardwareDescription;
import org.apache.flink.runtime.instance.InstanceID;
import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalListener;
import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService;
import org.apache.flink.runtime.memory.MemoryManager;
import org.apache.flink.runtime.memory.SharedResources;
import org.apache.flink.runtime.messages.Acknowledge;
import org.apache.flink.runtime.metrics.groups.ResourceProviderMetricGroup;
import org.apache.flink.runtime.providermanager.ConsumingItem;
import org.apache.flink.runtime.providermanager.ProviderManagerActions;
import org.apache.flink.runtime.providermanager.ResourceAllocationState;
import org.apache.flink.runtime.providermanager.UnresolvedProviderManagerLocation;
import org.apache.flink.runtime.registration.RegistrationConnectionListener;
import org.apache.flink.runtime.resourceconsumer.AllocatedSlotInfo;
import org.apache.flink.runtime.resourceconsumer.AllocatedSlotReport;
import org.apache.flink.runtime.resourceconsumer.ConsumeAttemptID;
import org.apache.flink.runtime.resourceconsumer.ConsumingMasterId;
import org.apache.flink.runtime.resourceconsumer.RCRPRegistrationRejection;
import org.apache.flink.runtime.resourceconsumer.RCRPRegistrationSuccess;
import org.apache.flink.runtime.resourceconsumer.ResourceConsumerGateway;
import org.apache.flink.runtime.resourceconsumer.ResourceManagerAddress;
import org.apache.flink.runtime.resourcemanager.ResourceManagerGateway;
import org.apache.flink.runtime.resourcemanager.ResourceManagerId;
import org.apache.flink.runtime.resourcemanager.ResourceProviderRegistration;
import org.apache.flink.runtime.resourceprovider.exceptions.ConsumingTaskSubmissionException;
import org.apache.flink.runtime.resourceprovider.exceptions.RegistrationTimeoutException;
import org.apache.flink.runtime.resourceprovider.exceptions.ResourceProviderException;
import org.apache.flink.runtime.resourceprovider.exceptions.SlotAllocationException;
import org.apache.flink.runtime.resourceprovider.exceptions.SlotOccupiedException;
import org.apache.flink.runtime.resourceprovider.slot.ConsumingTaskSlot;
import org.apache.flink.runtime.resourceprovider.slot.SlotActions;
import org.apache.flink.runtime.resourceprovider.slot.SlotAllocationSnapshot;
import org.apache.flink.runtime.resourceprovider.slot.SlotAllocationSnapshotPersistenceService;
import org.apache.flink.runtime.resourceprovider.slot.SlotNotActiveException;
import org.apache.flink.runtime.resourceprovider.slot.SlotNotFoundException;
import org.apache.flink.runtime.resourceprovider.slot.SlotOffer;
import org.apache.flink.runtime.resourceprovider.slot.TaskSlotTable;
import org.apache.flink.runtime.rest.messages.LogInfo;
import org.apache.flink.runtime.rest.messages.ThreadDumpInfo;
import org.apache.flink.runtime.rpc.FatalErrorHandler;
import org.apache.flink.runtime.rpc.RpcEndpoint;
import org.apache.flink.runtime.rpc.RpcService;
import org.apache.flink.runtime.rpc.RpcServiceUtils;
import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableList;
import org.apache.flink.shaded.guava30.com.google.common.collect.Sets;
import org.apache.flink.types.SerializableOptional;
import org.apache.flink.util.ERException;
import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.FlinkExpectedException;
import org.apache.flink.util.OptionalConsumer;
import org.apache.flink.util.Preconditions;
import org.apache.flink.util.StringUtils;
import org.apache.flink.util.concurrent.ExecutorThreadFactory;
import org.apache.flink.util.concurrent.FutureUtils;

/**
 * @类名: ResourceProvider
 * @说明: 资源提供者
 *
 * @author   leehom
 * @Date	 2024年2月2日 下午4:24:32
 * 修改记录：
 *
 * @see 	 
 */
/**
 * 资源提供者实
 * 负责提供多个消费者的资源.
 */
public class ResourceProvider extends RpcEndpoint implements ResourceProviderGateway {

    public static final String TASK_MANAGER_NAME = "taskmanager";

    /** The access to the leader election and retrieval services. */
    private final HighAvailabilityServices haServices;

    private final ResourceProviderServices resourceProviderServices;

    /** The resource provider configuration. */
    private final ResourceProviderConfiguration providerManagerConfiguration;

    /** The fatal error handler to use in case of a fatal error. */
    private final FatalErrorHandler fatalErrorHandler;

    private final TaskExecutorBlobService blobService;

    /** The address to metric query service on this Task Manager. */
    @Nullable private final String metricQueryServiceAddress;

    // --------- TaskManager services --------

    /** The connection information of this task manager. */
    private final UnresolvedProviderManagerLocation unresolvedTaskManagerLocation;

    private final ResourceProviderMetricGroup resourceProviderMetricGroup;

    /** Information provider for external resources. */
    private final ExternalResourceInfoProvider externalResourceInfoProvider;

    private final Executor ioExecutor;

    /** {@link MemoryManager} shared across all tasks. */
    private final SharedResources sharedResources;
    
    private final CommandListener commandListener;

    // --------- task slot allocation table -----------

    private final TaskSlotTable<ConsumingItem> taskSlotTable;

    /** consumer当前的资源slot offper*/
    private final Map<ConsumerID, UUID> currentSlotOfferPerConsumer = new HashMap<>();

    private final ConsumerTable consumerTable;

    private final ConsumerLeaderService consumerLeaderService;

    private final LeaderRetrievalService resourceManagerLeaderRetriever;

    private final SlotAllocationSnapshotPersistenceService slotAllocationSnapshotPersistenceService;

    // ------------------------------------------------------------------------

    private final HardwareDescription hardwareDescription;

    private final ResourceProviderMemoryConfiguration memoryConfiguration;

    private FileCache fileCache;

    /** The heartbeat manager for job manager in the task manager. */
    private final HeartbeatManager<AllocatedSlotReport, ResourceProviderToResourceConsumerHeartbeatPayload>
            resourceConsumerHeartbeatManager;

    /** The heartbeat manager for resource manager in the task manager. */
    private final HeartbeatManager<Void, ResourceProviderHeartbeatPayload>
            resourceManagerHeartbeatManager;

    // private final DelegationTokenReceiverRepository delegationTokenReceiverRepository;

    // --------- resource manager --------

    @Nullable private ResourceManagerAddress resourceManagerAddress;

    @Nullable private EstablishedResourceManagerConnection establishedResourceManagerConnection;

    @Nullable private ResourceProviderToResourceManagerConnection resourceManagerConnection;

    @Nullable private UUID currentRegistrationTimeoutId;

    private final ThreadInfoSampleService threadInfoSampleService;

    public ResourceProvider(
            RpcService rpcService,
            ResourceProviderConfiguration taskManagerConfiguration,
            HighAvailabilityServices haServices,
            ResourceProviderServices resourceProviderServices,
            ExternalResourceInfoProvider externalResourceInfoProvider,
            HeartbeatServices heartbeatServices,
            ResourceProviderMetricGroup resourceProviderMetricGroup,
            @Nullable String metricQueryServiceAddress,
            TaskExecutorBlobService blobService,
            FatalErrorHandler fatalErrorHandler,
            CommandListener commandListener) {

        super(rpcService, RpcServiceUtils.createRandomName(TASK_MANAGER_NAME));

        checkArgument(
                taskManagerConfiguration.getNumberSlots() > 0,
                "The number of slots has to be larger than 0.");

        this.providerManagerConfiguration = checkNotNull(taskManagerConfiguration);
        this.resourceProviderServices = checkNotNull(resourceProviderServices);
        this.haServices = checkNotNull(haServices);
        this.fatalErrorHandler = checkNotNull(fatalErrorHandler);
        // this.delegationTokenReceiverRepository = checkNotNull(delegationTokenReceiverRepository);
        this.resourceProviderMetricGroup = checkNotNull(resourceProviderMetricGroup);
        this.blobService = checkNotNull(blobService);
        this.metricQueryServiceAddress = metricQueryServiceAddress;
        this.externalResourceInfoProvider = checkNotNull(externalResourceInfoProvider);
        
        this.commandListener = commandListener;

        this.taskSlotTable = resourceProviderServices.getTaskSlotTable();
        this.consumerTable = resourceProviderServices.getConsumerTable();
        this.consumerLeaderService = resourceProviderServices.getJobLeaderService();
        this.unresolvedTaskManagerLocation =
                resourceProviderServices.getUnresolvedTaskManagerLocation();
        this.ioExecutor = resourceProviderServices.getIOExecutor();
        this.resourceManagerLeaderRetriever = haServices.getResourceManagerLeaderRetriever();

        this.hardwareDescription =
                HardwareDescription.extractFromSystem(resourceProviderServices.getManagedMemorySize());
        this.memoryConfiguration =
                ResourceProviderMemoryConfiguration.create(taskManagerConfiguration.getConfiguration());

        this.resourceManagerAddress = null;
        this.resourceManagerConnection = null;
        this.currentRegistrationTimeoutId = null;

        final ResourceID resourceId =
                resourceProviderServices.getUnresolvedTaskManagerLocation().getResourceID();
        this.resourceConsumerHeartbeatManager =
                createConsumerManagerHeartbeatManager(heartbeatServices, resourceId);
        this.resourceManagerHeartbeatManager =
                createResourceManagerHeartbeatManager(heartbeatServices, resourceId);

        ExecutorThreadFactory sampleThreadFactory =
                new ExecutorThreadFactory.Builder()
                        .setPoolName("flink-thread-info-sampler")
                        .build();
        ScheduledExecutorService sampleExecutor =
                Executors.newSingleThreadScheduledExecutor(sampleThreadFactory);
        this.threadInfoSampleService = new ThreadInfoSampleService(sampleExecutor);

        this.slotAllocationSnapshotPersistenceService =
                resourceProviderServices.getSlotAllocationSnapshotPersistenceService();

        this.sharedResources = resourceProviderServices.getSharedResources();
    }

    private HeartbeatManager<Void, ResourceProviderHeartbeatPayload>
            createResourceManagerHeartbeatManager(
                    HeartbeatServices heartbeatServices, ResourceID resourceId) {
        return heartbeatServices.createHeartbeatManager(
                resourceId, new ResourceManagerHeartbeatListener(), getMainThreadExecutor(), log);
    }

    private HeartbeatManager<AllocatedSlotReport, ResourceProviderToResourceConsumerHeartbeatPayload>
            createConsumerManagerHeartbeatManager(
                    HeartbeatServices heartbeatServices, ResourceID resourceId) {
        return heartbeatServices.createHeartbeatManager(
                resourceId, new ResourceConsumerHeartbeatListener(), getMainThreadExecutor(), log);
    }

    // TBD
    @Override
    public CompletableFuture<Boolean> canBeReleased() {
    	// 是否有资源被使用
        return CompletableFuture.completedFuture(false);
    }

    @Override
    public CompletableFuture<Collection<LogInfo>> requestLogList(Time timeout) {
        return CompletableFuture.supplyAsync(
                () -> {
                    final String logDir = providerManagerConfiguration.getTaskManagerLogDir();
                    if (logDir != null) {
                        final File[] logFiles = new File(logDir).listFiles();

                        if (logFiles == null) {
                            throw new CompletionException(
                                    new ERException(
                                            String.format(
                                                    "There isn't a log file in log dir %s.",
                                                    logDir)));
                        }

                        return Arrays.stream(logFiles)
                                .filter(File::isFile)
                                .map(
                                        logFile ->
                                                new LogInfo(
                                                        logFile.getName(),
                                                        logFile.length(),
                                                        logFile.lastModified()))
                                .collect(Collectors.toList());
                    }
                    return Collections.emptyList();
                },
                ioExecutor);
    }

    // ------------------------------------------------------------------------
    //  Life cycle
    // ------------------------------------------------------------------------

    @Override
    public void onStart() throws Exception {
        try {
            startResourceProviderServices();
        } catch (Throwable t) {
            final ResourceProviderException exception =
                    new ResourceProviderException(
                            String.format("Could not start the ResourceProvider %s", getAddress()), t);
            onFatalError(exception);
            throw exception;
        }

        startRegistrationTimeout();
    }

    private void startResourceProviderServices() throws Exception {
        try {
            // start by connecting to the ResourceManager
            resourceManagerLeaderRetriever.start(new ResourceManagerLeaderListener());

            // tell the task slot table who's responsible for the task slot actions
            taskSlotTable.start(new SlotActionsImpl(), getMainThreadExecutor());

            // start the job leader service
            consumerLeaderService.start(
                    getAddress(), getRpcService(), haServices, new ResourceConsumerLeaderListenerImpl());

            fileCache =
                    new FileCache(
                            providerManagerConfiguration.getTmpDirectories(),
                            blobService.getPermanentBlobService());

            tryLoadLocalAllocationSnapshots();
        } catch (Exception e) {
            handleStartTaskExecutorServicesException(e);
        }
    }

    private void handleStartTaskExecutorServicesException(Exception e) throws Exception {
        try {
            stopResourceProviderServices();
        } catch (Exception inner) {
            e.addSuppressed(inner);
        }

        throw e;
    }

    /** Called to shut down the TaskManager. The method closes all TaskManager services. */
    @Override
    public CompletableFuture<Void> onStop() {
        log.info("Stopping TaskExecutor {}.", getAddress());

        Throwable jobManagerDisconnectThrowable = null;

        FlinkExpectedException cause =
                new FlinkExpectedException("The TaskExecutor is shutting down.");

        closeResourceManagerConnection(cause);

        for (ConsumerTable.Consumer consumer : consumerTable.getConsumers()) {
            try {
                closeConsuming(consumer, cause);
            } catch (Throwable t) {
                jobManagerDisconnectThrowable =
                        ExceptionUtils.firstOrSuppressed(t, jobManagerDisconnectThrowable);
            }
        }

        Preconditions.checkState(consumerTable.isEmpty());

        final Throwable throwableBeforeTasksCompletion = jobManagerDisconnectThrowable;

        return FutureUtils.runAfterwards(taskSlotTable.closeAsync(), this::stopResourceProviderServices)
                .handle(
                        (ignored, throwable) -> {
                            handleOnStopException(throwableBeforeTasksCompletion, throwable);
                            return null;
                        });
    }

    private void handleOnStopException(
            Throwable throwableBeforeTasksCompletion, Throwable throwableAfterTasksCompletion) {
        final Throwable throwable;

        if (throwableBeforeTasksCompletion != null) {
            throwable =
                    ExceptionUtils.firstOrSuppressed(
                            throwableBeforeTasksCompletion, throwableAfterTasksCompletion);
        } else {
            throwable = throwableAfterTasksCompletion;
        }

        if (throwable != null) {
            throw new CompletionException(
                    new ERException("Error while shutting the TaskExecutor down.", throwable));
        } else {
            log.info("Stopped TaskExecutor {}.", getAddress());
        }
    }

    private void stopResourceProviderServices() throws Exception {
        Exception exception = null;

        try {
            threadInfoSampleService.close();
        } catch (Exception e) {
            exception = ExceptionUtils.firstOrSuppressed(e, exception);
        }

        try {
            consumerLeaderService.stop();
        } catch (Exception e) {
            exception = ExceptionUtils.firstOrSuppressed(e, exception);
        }

        try {
            resourceManagerLeaderRetriever.stop();
        } catch (Exception e) {
            exception = ExceptionUtils.firstOrSuppressed(e, exception);
        }

        try {
            resourceProviderServices.shutDown();
        } catch (Exception e) {
            exception = ExceptionUtils.firstOrSuppressed(e, exception);
        }

        try {
            fileCache.shutdown();
        } catch (Exception e) {
            exception = ExceptionUtils.firstOrSuppressed(e, exception);
        }

        // it will call close() recursively from the parent to children
        resourceProviderMetricGroup.close();

        ExceptionUtils.tryRethrowException(exception);
    }

    // ======================================================================
    //  RPC methods
    // ======================================================================
    
    // ----------------------------------------------------------------------
    // Heartbeat RPC
    // ----------------------------------------------------------------------
    @Override
    public CompletableFuture<Void> heartbeatFromResourceConsumer(
            ResourceID resourceID, AllocatedSlotReport allocatedSlotReport) {
        return resourceConsumerHeartbeatManager.requestHeartbeat(resourceID, allocatedSlotReport);
    }

    @Override
    public CompletableFuture<Void> heartbeatFromResourceManager(ResourceID resourceID) {
        return resourceManagerHeartbeatManager.requestHeartbeat(resourceID, null);
    }

    // ----------------------------------------------------------------------
    // Slot allocation RPCs
    // ----------------------------------------------------------------------
    // 消费者请求资源slot，资源提供者注册到资源管理器
    @Override
    public CompletableFuture<Acknowledge> requestSlot(
            final SlotID slotId, // 已注册到rm的slot
            final ConsumerID consumerId, // 申请的consumer
            final AllocationID allocationId,
            final ResourceProfile resourceProfile,
            final String targetAddress,
            final ResourceManagerId resourceManagerId,
            final Time timeout) {
        // TODO: Filter invalid requests from the resource manager by using the
        // instance/registration Id

        log.info(
                "Receive slot request {} for consumer {} from resource manager with leader id {}.",
                allocationId,
                consumerId,
                resourceManagerId);
        // 
        if (!isConnectedToResourceManager(resourceManagerId)) {
            final String message =
                    String.format(
                            "ResourceProvider is not connected to the resource manager %s.",
                            resourceManagerId);
            log.debug(message);
            return FutureUtils.completedExceptionally(new ResourceProviderException(message));
        }
        // 尝试持久分配快照
        tryPersistAllocationSnapshot(
                new SlotAllocationSnapshot(
                        slotId, consumerId, targetAddress, allocationId, resourceProfile));

        try {
            final boolean isConnected =
                    allocateSlotForConsumer(consumerId, slotId, allocationId, resourceProfile, targetAddress);

            if (isConnected) {
                offerSlotsToConsumerManager(consumerId);
            }

            return CompletableFuture.completedFuture(Acknowledge.get());
        } catch (SlotAllocationException e) {
            log.debug("Could not allocate slot for allocation id {}.", allocationId, e);
            return FutureUtils.completedExceptionally(e);
        }
    }

    private boolean allocateSlotForConsumer(
            ConsumerID consumerId,
            SlotID slotId,
            AllocationID allocationId,
            ResourceProfile resourceProfile,
            String targetAddress)
            throws SlotAllocationException {
    	// 
        allocateSlot(slotId, consumerId, allocationId, resourceProfile);
        // 
        final ConsumerTable.Consumer consumer;
        try {
            consumer =
                    consumerTable.getOrCreateConsumer(
                            consumerId, () -> registerNewJobAndCreateServices(consumerId, targetAddress));
        } catch (Exception e) {
            // free the allocated slot
            try {
                taskSlotTable.freeSlot(allocationId);
            } catch (SlotNotFoundException slotNotFoundException) {
                // slot no longer existent, this should actually never happen, because we've
                // just allocated the slot. So let's fail hard in this case!
                onFatalError(slotNotFoundException);
            }

            // sanity check
            if (!taskSlotTable.isSlotFree(slotId.getSlotNumber())) {
                onFatalError(new Exception("Could not free slot " + slotId));
            }

            throw new SlotAllocationException("Could not create new job.", e);
        }

        return consumer.isConnected();
    }

    private ResourceProviderConsumerServices registerNewJobAndCreateServices(
            ConsumerID consumerId, String targetAddress) throws Exception {
        consumerLeaderService.addJob(consumerId, targetAddress);
        final ConsumerPermanentBlobService permanentBlobService =
                blobService.getPermanentBlobService();
        permanentBlobService.registerJob(consumerId);

        return ResourceProviderConsumerServices.create(
                () -> permanentBlobService.releaseJob(consumerId));
    }

    // 
    private void allocateSlot(
            SlotID slotId, ConsumerID consumerId, AllocationID allocationId, ResourceProfile resourceProfile)
            throws SlotAllocationException {
        if (taskSlotTable.isSlotFree(slotId.getSlotNumber())) {
            if (taskSlotTable.allocateSlot(
                    slotId.getSlotNumber(),
                    consumerId,
                    allocationId,
                    resourceProfile,
                    providerManagerConfiguration.getSlotTimeout())) {
                log.info("Allocated slot for {}.", allocationId);
            } else {
                log.info("Could not allocate slot for {}.", allocationId);
                throw new SlotAllocationException("Could not allocate slot.");
            }
        } else if (!taskSlotTable.isAllocated(slotId.getSlotNumber(), consumerId, allocationId)) {
            final String message =
                    "The slot " + slotId + " has already been allocated for a different job.";

            log.info(message);

            final AllocationID allocationID =
                    taskSlotTable.getCurrentAllocation(slotId.getSlotNumber());
            throw new SlotOccupiedException(
                    message, allocationID, taskSlotTable.getOwningConsumer(allocationID));
        }
    }

    @Override
    public CompletableFuture<Acknowledge> freeSlot(
            AllocationID allocationId, Throwable cause, Time timeout) {
        freeSlotInternal(allocationId, cause);
        return CompletableFuture.completedFuture(Acknowledge.get());
    }

    @Override
    public void freeInactiveSlots(ConsumerID consumerId, Time timeout) {
        log.debug("Freeing inactive slots for job {}.", consumerId);

        // need a copy to prevent ConcurrentModificationExceptions
        final ImmutableList<ConsumingTaskSlot<ConsumingItem>> inactiveSlots =
                ImmutableList.copyOf(taskSlotTable.getAllocatedSlots(consumerId));
        for (ConsumingTaskSlot<ConsumingItem> slot : inactiveSlots) {
            freeSlotInternal(
                    slot.getAllocationId(),
                    new ERException("Slot was re-claimed by resource manager."));
        }
    }

    @Override
    public CompletableFuture<TransientBlobKey> requestFileUploadByType(
            FileType fileType, Time timeout) {
        final String filePath;
        switch (fileType) {
            case LOG:
                filePath = providerManagerConfiguration.getTaskManagerLogPath();
                break;
            case STDOUT:
                filePath = providerManagerConfiguration.getTaskManagerStdoutPath();
                break;
            default:
                filePath = null;
        }
        return requestFileUploadByFilePath(filePath, fileType.toString());
    }

    @Override
    public CompletableFuture<TransientBlobKey> requestFileUploadByName(
            String fileName, Time timeout) {
        final String filePath;
        final String logDir = providerManagerConfiguration.getTaskManagerLogDir();
        if (StringUtils.isNullOrWhitespaceOnly(logDir)
                || StringUtils.isNullOrWhitespaceOnly(fileName)) {
            filePath = null;
        } else {
            filePath = new File(logDir, new File(fileName).getName()).getPath();
        }
        return requestFileUploadByFilePath(filePath, fileName);
    }

    @Override
    public CompletableFuture<SerializableOptional<String>> requestMetricQueryServiceAddress(
            Time timeout) {
        return CompletableFuture.completedFuture(
                SerializableOptional.ofNullable(metricQueryServiceAddress));
    }

    // ----------------------------------------------------------------------
    // Disconnection RPCs
    // ----------------------------------------------------------------------

    @Override
    public void disconnectResourceConsumer(ConsumerID consumerId, Exception cause) {
        consumerTable.getConnection(consumerId)
                .ifPresent(
                        jobManagerConnection ->
                                disconnectAndTryReconnectToResourceConsumer(jobManagerConnection, cause));
    }

    private void disconnectAndTryReconnectToResourceConsumer(
            ConsumerTable.Connection jobManagerConnection, Exception cause) {
        disconnectResourceConsumerConnection(jobManagerConnection, cause);
        consumerLeaderService.reconnect(jobManagerConnection.getConsumerId());
    }

    @Override
    public void disconnectResourceManager(Exception cause) {
        if (isRunning()) {
            reconnectToResourceManager(cause);
        }
    }

    // ----------------------------------------------------------------------
    // Other RPCs
    // ----------------------------------------------------------------------

    @Override
    public CompletableFuture<ThreadDumpInfo> requestThreadDump(Time timeout) {
        int stacktraceMaxDepth =
                providerManagerConfiguration
                        .getConfiguration()
                        .get(ClusterOptions.THREAD_DUMP_STACKTRACE_MAX_DEPTH);
        return CompletableFuture.completedFuture(ThreadDumpInfo.dumpAndCreate(stacktraceMaxDepth));
    }

    //
//    @Override
//    public CompletableFuture<Acknowledge> updateDelegationTokens(
//            ResourceManagerId resourceManagerId, byte[] tokens) {
//        log.info(
//                "Receive update delegation tokens from resource manager with leader id {}.",
//                resourceManagerId);
//
//        if (!isConnectedToResourceManager(resourceManagerId)) {
//            final String message =
//                    String.format(
//                            "TaskManager is not connected to the resource manager %s.",
//                            resourceManagerId);
//            log.debug(message);
//            return FutureUtils.completedExceptionally(new ResourceProviderException(message));
//        }
//
//        try {
//            delegationTokenReceiverRepository.onNewTokensObtained(tokens);
//            return CompletableFuture.completedFuture(Acknowledge.get());
//        } catch (Throwable t) {
//            log.error("Could not update delegation tokens.", t);
//            ExceptionUtils.rethrowIfFatalError(t);
//            return FutureUtils.completedExceptionally(t);
//        }
//    }

    // ------------------------------------------------------------------------
    //  Internal resource manager connection methods
    // ------------------------------------------------------------------------

    private void notifyOfNewResourceManagerLeader(
            String newLeaderAddress, ResourceManagerId newResourceManagerId) {
    	// 获取信资源管理器地址
        resourceManagerAddress =
                createResourceManagerAddress(newLeaderAddress, newResourceManagerId);
        reconnectToResourceManager(
                new ERException(
                        String.format(
                                "ResourceManager leader changed to new address %s",
                                resourceManagerAddress)));
    }

    @Nullable
    private ResourceManagerAddress createResourceManagerAddress(
            @Nullable String newLeaderAddress, @Nullable ResourceManagerId newResourceManagerId) {
        if (newLeaderAddress == null) {
            return null;
        } else {
            assert (newResourceManagerId != null);
            return new ResourceManagerAddress(newLeaderAddress, newResourceManagerId);
        }
    }

    private void reconnectToResourceManager(Exception cause) {
        closeResourceManagerConnection(cause);
        startRegistrationTimeout();
        tryConnectToResourceManager();
    }

    private void tryConnectToResourceManager() {
        if (resourceManagerAddress != null) {
            connectToResourceManager();
        }
    }

    // 注册到资源管理器
    private void connectToResourceManager() {
        assert (resourceManagerAddress != null);
        assert (establishedResourceManagerConnection == null);
        assert (resourceManagerConnection == null);

        log.info("Connecting to ResourceManager {}.", resourceManagerAddress);

        final ResourceProviderRegistration resourceProviderRegistration =
                new ResourceProviderRegistration(
                        getAddress(),
                        getResourceID(),
                        JMXService.getPort().orElse(-1),
                        hardwareDescription,
                        memoryConfiguration,
                        providerManagerConfiguration.getDefaultSlotResourceProfile(),
                        providerManagerConfiguration.getTotalResourceProfile(),
                        unresolvedTaskManagerLocation.getNodeId());

        resourceManagerConnection =
                new ResourceProviderToResourceManagerConnection(
                        log,
                        getRpcService(),
                        providerManagerConfiguration.getRetryingRegistrationConfiguration(),
                        resourceManagerAddress.getAddress(),
                        resourceManagerAddress.getResourceManagerId(),
                        getMainThreadExecutor(),
                        new ResourceManagerRegistrationListener(),
                        resourceProviderRegistration);
        resourceManagerConnection.start();
    }

    //
    private void establishResourceManagerConnection(
            ResourceManagerGateway resourceManagerGateway,
            ResourceID resourceManagerResourceId,
            InstanceID taskExecutorRegistrationId,
            ClusterInformation clusterInformation) {
    	// 报告资源
        final CompletableFuture<Acknowledge> slotReportResponseFuture =
                resourceManagerGateway.sendSlotReport(
                        getResourceID(),
                        taskExecutorRegistrationId,
                        taskSlotTable.createSlotReport(getResourceID()),
                        providerManagerConfiguration.getRpcTimeout());
        // 异步完成
        slotReportResponseFuture.whenCompleteAsync(
                (acknowledge, throwable) -> {
                    if (throwable != null) {
                        reconnectToResourceManager(
                                new ResourceProviderException(
                                        "Failed to send initial slot report to ResourceManager.",
                                        throwable));
                    }
                },
                getMainThreadExecutor());

        // monitor the resource manager as heartbeat target
        resourceManagerHeartbeatManager.monitorTarget(
                resourceManagerResourceId,
                new ResourceManagerHeartbeatReceiver(resourceManagerGateway));

        // set the propagated blob server address
        final InetSocketAddress blobServerAddress =
                new InetSocketAddress(
                        clusterInformation.getBlobServerHostname(),
                        clusterInformation.getBlobServerPort());

        blobService.setBlobServerAddress(blobServerAddress);
        // 持有连接，用于后续关闭
        establishedResourceManagerConnection =
                new EstablishedResourceManagerConnection(
                        resourceManagerGateway,
                        resourceManagerResourceId,
                        taskExecutorRegistrationId);

        stopRegistrationTimeout();
    }

    private void closeResourceManagerConnection(Exception cause) {
        if (establishedResourceManagerConnection != null) {
            final ResourceID resourceManagerResourceId =
                    establishedResourceManagerConnection.getResourceManagerResourceId();

            log.info(
                    "Close ResourceManager connection {}.",
                    resourceManagerResourceId,
                    ExceptionUtils.returnExceptionIfUnexpected(cause.getCause()));
            ExceptionUtils.logExceptionIfExcepted(cause.getCause(), log);
            resourceManagerHeartbeatManager.unmonitorTarget(resourceManagerResourceId);

            ResourceManagerGateway resourceManagerGateway =
                    establishedResourceManagerConnection.getResourceManagerGateway();
            resourceManagerGateway.disconnectTaskManager(getResourceID(), cause);

            establishedResourceManagerConnection = null;
            
        }

        if (resourceManagerConnection != null) {
            if (!resourceManagerConnection.isConnected()) {
                if (log.isDebugEnabled()) {
                    log.debug(
                            "Terminating registration attempts towards ResourceManager {}.",
                            resourceManagerConnection.getTargetAddress(),
                            cause);
                } else {
                    log.info(
                            "Terminating registration attempts towards ResourceManager {}.",
                            resourceManagerConnection.getTargetAddress());
                }
            }

            resourceManagerConnection.close();
            resourceManagerConnection = null;
        }
    }

    private void startRegistrationTimeout() {
        final Duration maxRegistrationDuration =
                providerManagerConfiguration.getMaxRegistrationDuration();

        if (maxRegistrationDuration != null) {
            final UUID newRegistrationTimeoutId = UUID.randomUUID();
            currentRegistrationTimeoutId = newRegistrationTimeoutId;
            scheduleRunAsync(
                    () -> registrationTimeout(newRegistrationTimeoutId), maxRegistrationDuration);
        }
    }

    private void stopRegistrationTimeout() {
        currentRegistrationTimeoutId = null;
    }

    private void registrationTimeout(@Nonnull UUID registrationTimeoutId) {
        if (registrationTimeoutId.equals(currentRegistrationTimeoutId)) {
            final Duration maxRegistrationDuration =
                    providerManagerConfiguration.getMaxRegistrationDuration();

            onFatalError(
                    new RegistrationTimeoutException(
                            String.format(
                                    "Could not register at the ResourceManager within the specified maximum "
                                            + "registration duration %s. This indicates a problem with this instance. Terminating now.",
                                    maxRegistrationDuration)));
        }
    }

    // ------------------------------------------------------------------------
    //  Internal provider manager connection methods
    // ------------------------------------------------------------------------

    private void offerSlotsToConsumerManager(final ConsumerID consumerId) {
        consumerTable.getConnection(consumerId).ifPresent(this::internalOfferSlotsToConsumerManager);
    }

    private void internalOfferSlotsToConsumerManager(ConsumerTable.Connection resourceConsumerConnection) {
        final ConsumerID consumerId = resourceConsumerConnection.getConsumerId();

        if (taskSlotTable.hasAllocatedSlots(consumerId)) {
            log.info("Offer reserved slots to the leader of consumer {}.", consumerId);

            final ResourceConsumerGateway jobMasterGateway = resourceConsumerConnection.getResourceConsumerGateway();

            final Iterator<ConsumingTaskSlot<ConsumingItem>> reservedSlotsIterator =
                    							taskSlotTable.getAllocatedSlots(consumerId);
            final ConsumingMasterId consumingMasterId = resourceConsumerConnection.getConsumingMasterId();

            final Collection<SlotOffer> reservedSlots = new HashSet<>(2);

            while (reservedSlotsIterator.hasNext()) {
                SlotOffer offer = reservedSlotsIterator.next().generateSlotOffer();
                reservedSlots.add(offer);
            }

            final UUID slotOfferId = UUID.randomUUID();
            currentSlotOfferPerConsumer.put(consumerId, slotOfferId);

            CompletableFuture<Collection<SlotOffer>> acceptedSlotsFuture =
                    jobMasterGateway.offerSlots(
                            getResourceID(),
                            reservedSlots,
                            providerManagerConfiguration.getRpcTimeout());

            acceptedSlotsFuture.whenCompleteAsync(
                    handleAcceptedSlotOffers(
                            consumerId, jobMasterGateway, consumingMasterId, reservedSlots, slotOfferId),
                    getMainThreadExecutor());
        } else {
            log.debug("There are no unassigned slots for the job {}.", consumerId);
        }
    }

    //
    @Nonnull
    private BiConsumer<Iterable<SlotOffer>, Throwable> handleAcceptedSlotOffers(
            ConsumerID consumerId,
            ResourceConsumerGateway resourceConsumerGateway,
            ConsumingMasterId consumingMasterId,
            Collection<SlotOffer> offeredSlots,
            UUID offerId) {
        return (Iterable<SlotOffer> acceptedSlots, Throwable throwable) -> {
            // check if this is the latest offer
            if (!offerId.equals(currentSlotOfferPerConsumer.get(consumerId))) {
                // If this offer is outdated then it can be safely ignored.
                // If the response for a given slot is identical in both offers (accepted/rejected),
                // then this is naturally the case since the end-result is the same.
                // If the responses differ, then there are 2 cases to consider:
                // 1) initially rejected, later accepted
                //   This can happen when the resource requirements of a job increases between
                //   offers.
                //   In this case the first response MUST be ignored, so that
                //   the slot can be properly activated when the second response arrives.
                // 2) initially accepted, later rejected
                //   This can happen when the resource requirements of a job decrease between
                //   offers.
                //   In this case the first response MAY be ignored, because the job no longer
                //   requires the slot (and already has initiated steps to free it) and we can thus
                //   assume that any in-flight task submissions are no longer relevant for the job
                //   execution.

                log.debug(
                        "Discard slot offer response since there is a newer offer for the job {}.",
                        consumerId);
                return;
            }
            //
            if (throwable != null) {
                if (throwable instanceof TimeoutException) {
                    log.info(
                            "Slot offering to JobManager did not finish in time. Retrying the slot offering.");
                    // We ran into a timeout. Try again.
                    offerSlotsToConsumerManager(consumerId);
                } else {
                    log.warn(
                            "Slot offering to JobManager failed. Freeing the slots "
                                    + "and returning them to the ResourceManager.",
                            throwable);

                    // We encountered an exception. Free the slots and return them to the RM.
                    for (SlotOffer reservedSlot : offeredSlots) {
                        freeSlotInternal(reservedSlot.getAllocationId(), throwable);
                    }
                }
            } else {
                // check if the response is still valid
                if (isResourceConsumerConnectionValid(consumerId, consumingMasterId)) {
                    // mark accepted slots active
                    for (SlotOffer acceptedSlot : acceptedSlots) {
                        final AllocationID allocationId = acceptedSlot.getAllocationId();
                        try {
                            if (!taskSlotTable.markSlotActive(allocationId)) {
                                // the slot is either free or releasing at the moment
                                final String message =
                                        "Could not mark slot " + allocationId + " active.";
                                log.debug(message);
                                resourceConsumerGateway.failSlot(
                                        getResourceID(), allocationId, new ERException(message));
                            }
                        } catch (SlotNotFoundException e) {
                            final String message =
                                    "Could not mark slot " + allocationId + " active.";
                            resourceConsumerGateway.failSlot(
                                    getResourceID(), allocationId, new ERException(message));
                        }
                        // 
                        offeredSlots.remove(acceptedSlot);
                    }

                    final Exception e = new Exception("The slot was rejected by the JobManager.");

                    for (SlotOffer rejectedSlot : offeredSlots) {
                        freeSlotInternal(rejectedSlot.getAllocationId(), e);
                    }
                } else {
                    // discard the response since there is a new leader for the job
                    log.debug(
                            "Discard slot offer response since there is a new leader "
                                    + "for the job {}.",
                            consumerId);
                }
            }
        };
    }

    private void establishResourceConsumerConnection(
            ConsumerTable.Consumer consumer,
            final ResourceConsumerGateway resourceConsumerGateway,
            RCRPRegistrationSuccess registrationSuccess) {

        final ConsumerID consumerId = consumer.getConsumerId();
        final Optional<ConsumerTable.Connection> connection = consumer.asConnection();

        if (connection.isPresent()) {
            ConsumerTable.Connection oldJobManagerConnection = connection.get();

            if (Objects.equals(
                    oldJobManagerConnection.getConsumingMasterId(), resourceConsumerGateway.getFencingToken())) {
                // we already are connected to the given job manager
                log.debug(
                        "Ignore JobManager gained leadership message for {} because we are already connected to it.",
                        resourceConsumerGateway.getFencingToken());
                return;
            } else {
                disconnectResourceConsumerConnection(
                        oldJobManagerConnection,
                        new Exception("Found new job leader for job id " + consumerId + '.'));
            }
        }

        log.info("Establish JobManager connection for job {}.", consumerId);

        ResourceID consumerManagerResourceID = registrationSuccess.getResourceID();

        final ConsumerTable.Connection establishedConnection =
                associateWithResourceConsumer(consumer, consumerManagerResourceID, resourceConsumerGateway);

        // monitor the job manager as heartbeat target
        resourceConsumerHeartbeatManager.monitorTarget(
                consumerManagerResourceID, new ConsumerManagerHeartbeatReceiver(resourceConsumerGateway));

        internalOfferSlotsToConsumerManager(establishedConnection);
    }

    private void closeConsuming(ConsumerTable.Consumer consumer, Exception cause) {
        consumer.asConnection()
                .ifPresent(
                        consumerManagerConnection ->
                                disconnectResourceConsumerConnection(consumerManagerConnection, cause));

        consumer.close();
    }

    private void disconnectResourceConsumerConnection(
            ConsumerTable.Connection ConsumerManagerConnection, Exception cause) {
        final ConsumerID consumerId = ConsumerManagerConnection.getConsumerId();
        log.info(
                "Close ConsumerManager connection for consumer {}.",
                consumerId,
                ExceptionUtils.returnExceptionIfUnexpected(cause.getCause()));
        ExceptionUtils.logExceptionIfExcepted(cause.getCause(), log);

        // 1. fail tasks running under this ConsumerID
        Iterator<ConsumingItem> tasks = taskSlotTable.getTasks(consumerId);

        final ERException failureCause =
                new ERException(
                        String.format("Disconnect from ConsumerManager responsible for %s.", consumerId),
                        cause);

        // 2. Move the active slots to state allocated (possible to time out again)
        Set<AllocationID> activeSlotAllocationIDs =
                taskSlotTable.getActiveTaskSlotAllocationIdsPerConsumer(consumerId);

        final ERException freeingCause =
                new ERException("Slot could not be marked inactive.");

        for (AllocationID activeSlotAllocationID : activeSlotAllocationIDs) {
            try {
                if (!taskSlotTable.markSlotInactive(
                        activeSlotAllocationID, providerManagerConfiguration.getSlotTimeout())) {
                    freeSlotInternal(activeSlotAllocationID, freeingCause);
                }
            } catch (SlotNotFoundException e) {
                log.debug("Could not mark the slot {} inactive.", activeSlotAllocationID, e);
            }
        }

        // 3. Disassociate from the Resource Consumer
        try {
            resourceConsumerHeartbeatManager.unmonitorTarget(ConsumerManagerConnection.getResourceId());
            disassociateFromConsumerManager(ConsumerManagerConnection, cause);
        } catch (IOException e) {
            log.warn(
                    "Could not properly disassociate from ConsumerManager {}.",
                    ConsumerManagerConnection.getResourceConsumerGateway().getAddress(),
                    e);
        }

        ConsumerManagerConnection.disconnect();
    }

    // 
    private ConsumerTable.Connection associateWithResourceConsumer(
            ConsumerTable.Consumer consumer, ResourceID resourceID, ResourceConsumerGateway jobMasterGateway) {
        checkNotNull(resourceID);
        checkNotNull(jobMasterGateway);

        ProviderManagerActions taskManagerActions = new ProviderManagerActionsImpl(jobMasterGateway);

        registerQueryableState(consumer.getConsumerId(), jobMasterGateway);

        return consumer.connect(
                resourceID,
                jobMasterGateway,
                taskManagerActions);
    }

    private void disassociateFromConsumerManager(
            ConsumerTable.Connection jobManagerConnection, Exception cause) throws IOException {
        checkNotNull(jobManagerConnection);

        final ConsumerID jobId = jobManagerConnection.getConsumerId();

        ResourceConsumerGateway jobManagerGateway = jobManagerConnection.getResourceConsumerGateway();
        jobManagerGateway.disconnectTaskManager(getResourceID(), cause);
    }

    private void handleRejectedJobManagerConnection(
            ConsumerID jobId, String targetAddress, RCRPRegistrationRejection rejection) {
        log.info(
                "The ConsumerManager under {} rejected the registration for consumer {}: {}. Releasing all job related resources.",
                targetAddress,
                jobId,
                rejection.getReason());

        releaseConsumerResources(
                jobId,
                new ERException(
                        String.format("JobManager %s has rejected the registration.", jobId)));
    }

    private void releaseConsumerResources(ConsumerID consumerId, Exception cause) {
        log.debug("Releasing resources for consumer {}.", consumerId, cause);

        // free slots
        final Set<AllocationID> allocationIds = taskSlotTable.getAllocationIdsPerConsumer(consumerId);

        if (!allocationIds.isEmpty()) {
            for (AllocationID allocationId : allocationIds) {
                freeSlotInternal(allocationId, cause);
            }
        }

        consumerLeaderService.removeJob(consumerId);
        consumerTable.getConsumer(consumerId)
                .ifPresent(
                        job -> {
                            closeConsuming(job, cause);
                        });
        resourceProviderMetricGroup.removeJobMetricsGroup(consumerId);
        currentSlotOfferPerConsumer.remove(consumerId);
    }

    private void registerQueryableState(ConsumerID jobId, ResourceConsumerGateway jobMasterGateway) {

    }

    // ------------------------------------------------------------------------
    //  Internal task methods
    // ------------------------------------------------------------------------

    private void failTask(final ExecutionID executionId, final Throwable cause) {
        final ConsumingItem task = taskSlotTable.getTask(executionId);
        
    }

    private void updateTaskExecutionState(
            final ResourceConsumerGateway consumerMasterGateway, final ResourceAllocationState taskExecutionState) {

    }

    private void unregisterTaskAndNotifyFinalState(
            final ResourceConsumerGateway jobMasterGateway, final ConsumeAttemptID executionAttemptID) {

    }

    private void freeSlotInternal(AllocationID allocationId, Throwable cause) {
        checkNotNull(allocationId);

        // only respond to freeing slots when not shutting down to avoid freeing slot allocation
        // information
        if (isRunning()) {
            log.debug(
                    "Free slot with allocation id {} because: {}",
                    allocationId,
                    cause.getMessage());

            try {
                final ConsumerID consumerId = taskSlotTable.getOwningConsumer(allocationId);

                final int slotIndex = taskSlotTable.freeSlot(allocationId, cause);

                slotAllocationSnapshotPersistenceService.deleteAllocationSnapshot(slotIndex);

                if (slotIndex != -1) {

                    if (isConnectedToResourceManager()) {
                        // the slot was freed. Tell the RM about it
                        ResourceManagerGateway resourceManagerGateway =
                                establishedResourceManagerConnection.getResourceManagerGateway();

                        resourceManagerGateway.notifySlotAvailable(
                                establishedResourceManagerConnection
                                        .getTaskExecutorRegistrationId(),
                                new SlotID(getResourceID(), slotIndex),
                                allocationId);
                    }

                    if (consumerId != null) {
                        closeJobManagerConnectionIfNoAllocatedResources(consumerId);
                    }
                }
            } catch (SlotNotFoundException e) {
                log.debug("Could not free slot for allocation id {}.", allocationId, e);
            }

        } else {
            log.debug(
                    "Ignoring the freeing of slot {} because the TaskExecutor is shutting down.",
                    allocationId);
        }
    }

    private void closeJobManagerConnectionIfNoAllocatedResources(ConsumerID consumerId) {
        // check whether we still have allocated slots for the same job
        if (taskSlotTable.getAllocationIdsPerConsumer(consumerId).isEmpty()) {
            // we can remove the job from the job leader service

            final FlinkExpectedException cause =
                    new FlinkExpectedException(
                            "TaskExecutor "
                                    + getAddress()
                                    + " has no more allocated slots for job "
                                    + consumerId
                                    + '.');

            releaseConsumerResources(consumerId, cause);
        }
    }

    private void timeoutSlot(AllocationID allocationId, UUID ticket) {
        checkNotNull(allocationId);
        checkNotNull(ticket);

        if (taskSlotTable.isValidTimeout(allocationId, ticket)) {
            freeSlotInternal(
                    allocationId, new Exception("The slot " + allocationId + " has timed out."));
        } else {
            log.debug(
                    "Received an invalid timeout for allocation id {} with ticket {}.",
                    allocationId,
                    ticket);
        }
    }

    /**
     * Syncs the TaskExecutor's view on its allocated slots with the JobMaster's view. Slots which
     * are no longer reported by the JobMaster are being freed. Slots which the JobMaster thinks it
     * still owns but which are no longer allocated to it will be failed via {@link
     * JobMasterGateway#failSlot}.
     *
     * @param jobMasterGateway jobMasterGateway to talk to the connected job master
     * @param allocatedSlotReport represents the JobMaster's view on the current slot allocation
     *     state
     */
    private void syncSlotsWithSnapshotFromJobMaster(
    		ResourceConsumerGateway jobMasterGateway, AllocatedSlotReport allocatedSlotReport) {
        failNoLongerAllocatedSlots(allocatedSlotReport, jobMasterGateway);
        freeNoLongerUsedSlots(allocatedSlotReport);
    }

    private void failNoLongerAllocatedSlots(
            AllocatedSlotReport allocatedSlotReport, ResourceConsumerGateway jobMasterGateway) {
        for (AllocatedSlotInfo allocatedSlotInfo : allocatedSlotReport.getAllocatedSlotInfos()) {
            final AllocationID allocationId = allocatedSlotInfo.getAllocationId();
            if (!taskSlotTable.isAllocated(
                    allocatedSlotInfo.getSlotIndex(),
                    allocatedSlotReport.getJobId(),
                    allocationId)) {
                jobMasterGateway.failSlot(
                        getResourceID(),
                        allocationId,
                        new ERException(
                                String.format(
                                        "Slot %s on TaskExecutor %s is not allocated by job %s.",
                                        allocatedSlotInfo.getSlotIndex(),
                                        getResourceID().getStringWithMetadata(),
                                        allocatedSlotReport.getJobId())));
            }
        }
    }

    private void freeNoLongerUsedSlots(AllocatedSlotReport allocatedSlotReport) {
        final Set<AllocationID> activeSlots =
                taskSlotTable.getActiveTaskSlotAllocationIdsPerConsumer(allocatedSlotReport.getJobId());
        final Set<AllocationID> reportedSlots =
                allocatedSlotReport.getAllocatedSlotInfos().stream()
                        .map(AllocatedSlotInfo::getAllocationId)
                        .collect(Collectors.toSet());

        final Sets.SetView<AllocationID> difference = Sets.difference(activeSlots, reportedSlots);

        for (AllocationID allocationID : difference) {
            freeSlotInternal(
                    allocationID,
                    new ERException(
                            String.format(
                                    "%s is no longer allocated by job %s.",
                                    allocationID, allocatedSlotReport.getJobId())));
        }
    }

    private void tryPersistAllocationSnapshot(SlotAllocationSnapshot slotAllocationSnapshot) {
        try {
            slotAllocationSnapshotPersistenceService.persistAllocationSnapshot(
                    slotAllocationSnapshot);
        } catch (IOException e) {
            log.debug("Cannot persist the slot allocation snapshot {}.", slotAllocationSnapshot, e);
        }
    }

    /**
     * This method tries to repopulate the {@link ConsumerTable} and {@link TaskSlotTable} from the local
     * filesystem in a best-effort manner.
     */
    private void tryLoadLocalAllocationSnapshots() {
        Collection<SlotAllocationSnapshot> slotAllocationSnapshots =
                slotAllocationSnapshotPersistenceService.loadAllocationSnapshots();

        log.debug("Recovered slot allocation snapshots {}.", slotAllocationSnapshots);

        final Set<AllocationID> allocatedSlots = new HashSet<>();
        for (SlotAllocationSnapshot slotAllocationSnapshot : slotAllocationSnapshots) {
            try {
                allocateSlotForConsumer(
                        slotAllocationSnapshot.getConsumerId(),
                        slotAllocationSnapshot.getSlotID(),
                        slotAllocationSnapshot.getAllocationId(),
                        slotAllocationSnapshot.getResourceProfile(),
                        slotAllocationSnapshot.getConsumerTargetAddress());

            } catch (SlotAllocationException e) {
                log.debug("Cannot reallocate restored slot {}.", slotAllocationSnapshot, e);
            }

            allocatedSlots.add(slotAllocationSnapshot.getAllocationId());
        }

    }

    // ------------------------------------------------------------------------
    //  Internal utility methods
    // ------------------------------------------------------------------------

    private boolean isConnectedToResourceManager() {
        return establishedResourceManagerConnection != null;
    }

    private boolean isConnectedToResourceManager(ResourceManagerId resourceManagerId) {
        return establishedResourceManagerConnection != null
                && resourceManagerAddress != null
                && resourceManagerAddress.getResourceManagerId().equals(resourceManagerId);
    }

    private boolean isResourceConsumerConnectionValid(ConsumerID consumerId, ConsumingMasterId consumingMasterId) {
        return consumerTable.getConnection(consumerId)
                .map(jmConnection -> Objects.equals(jmConnection.getConsumingMasterId(), consumingMasterId))
                .orElse(false);
    }

    private CompletableFuture<TransientBlobKey> requestFileUploadByFilePath(
            String filePath, String fileTag) {
        log.debug("Received file upload request for file {}", fileTag);
        if (!StringUtils.isNullOrWhitespaceOnly(filePath)) {
            return CompletableFuture.supplyAsync(
                    () -> {
                        final File file = new File(filePath);
                        if (file.exists()) {
                            try {
                                return putTransientBlobStream(new FileInputStream(file), fileTag)
                                        .get();
                            } catch (Exception e) {
                                log.debug("Could not upload file {}.", fileTag, e);
                                throw new CompletionException(
                                        new ERException(
                                                "Could not upload file " + fileTag + '.', e));
                            }
                        } else {
                            log.debug(
                                    "The file {} does not exist on the TaskExecutor {}.",
                                    fileTag,
                                    getResourceID().getStringWithMetadata());
                            throw new CompletionException(
                                    new ERException(
                                            "The file "
                                                    + fileTag
                                                    + " does not exist on the TaskExecutor."));
                        }
                    },
                    ioExecutor);
        } else {
            log.debug(
                    "The file {} is unavailable on the TaskExecutor {}.",
                    fileTag,
                    getResourceID().getStringWithMetadata());
            return FutureUtils.completedExceptionally(
                    new ERException(
                            "The file " + fileTag + " is not available on the TaskExecutor."));
        }
    }

    private CompletableFuture<TransientBlobKey> putTransientBlobStream(
            InputStream inputStream, String fileTag) {
        final TransientBlobService transientBlobService =
                blobService.getTransientBlobService();
        final TransientBlobKey transientBlobKey;

        try {
            transientBlobKey = transientBlobService.putTransient(inputStream);
        } catch (IOException e) {
            log.debug("Could not upload file {}.", fileTag, e);
            return FutureUtils.completedExceptionally(
                    new ERException("Could not upload file " + fileTag + '.', e));
        }
        return CompletableFuture.completedFuture(transientBlobKey);
    }

    // ------------------------------------------------------------------------
    //  Properties
    // ------------------------------------------------------------------------

    public ResourceID getResourceID() {
        return unresolvedTaskManagerLocation.getResourceID();
    }

    // ------------------------------------------------------------------------
    //  Error Handling
    // ------------------------------------------------------------------------

    /**
     * Notifies the TaskExecutor that a fatal error has occurred and it cannot proceed.
     *
     * @param t The exception describing the fatal error
     */
    void onFatalError(final Throwable t) {
        try {
            log.error("Fatal error occurred in TaskExecutor {}.", getAddress(), t);
        } catch (Throwable ignored) {
        }

        // The fatal error handler implementation should make sure that this call is non-blocking
        fatalErrorHandler.onFatalError(t);
    }

    // ------------------------------------------------------------------------
    //  Access to fields for testing
    // ------------------------------------------------------------------------

    @VisibleForTesting
    ResourceProviderToResourceManagerConnection getResourceManagerConnection() {
        return resourceManagerConnection;
    }

    @VisibleForTesting
    HeartbeatManager<Void, ResourceProviderHeartbeatPayload> getResourceManagerHeartbeatManager() {
        return resourceManagerHeartbeatManager;
    }

    // ------------------------------------------------------------------------
    //  Utility classes
    // ------------------------------------------------------------------------

    private static final class ConsumerManagerHeartbeatReceiver
            extends HeartbeatReceiver<ResourceProviderToResourceConsumerHeartbeatPayload> {
        private final ResourceConsumerGateway resourceConsumerGateway;

        private ConsumerManagerHeartbeatReceiver(ResourceConsumerGateway jobMasterGateway) {
            this.resourceConsumerGateway = jobMasterGateway;
        }

        @Override
        public CompletableFuture<Void> receiveHeartbeat(
                ResourceID resourceID, ResourceProviderToResourceConsumerHeartbeatPayload payload) {
            return resourceConsumerGateway.heartbeatFromTaskManager(resourceID, payload);
        }
    }

    private static final class ResourceManagerHeartbeatReceiver
            extends HeartbeatReceiver<ResourceProviderHeartbeatPayload> {
        private final ResourceManagerGateway resourceManagerGateway;

        private ResourceManagerHeartbeatReceiver(ResourceManagerGateway resourceManagerGateway) {
            this.resourceManagerGateway = resourceManagerGateway;
        }

        @Override
        public CompletableFuture<Void> receiveHeartbeat(
                ResourceID resourceID, ResourceProviderHeartbeatPayload heartbeatPayload) {
            return resourceManagerGateway.heartbeatFromTaskManager(resourceID, heartbeatPayload);
        }
    }

    /** The listener for leader changes of the resource manager. */
    private final class ResourceManagerLeaderListener implements LeaderRetrievalListener {

        @Override
        public void notifyLeaderAddress(final String leaderAddress, final UUID leaderSessionID) {
            runAsync(
                    () ->
                            notifyOfNewResourceManagerLeader(
                                    leaderAddress,
                                    ResourceManagerId.fromUuidOrNull(leaderSessionID)));
        }

        @Override
        public void handleError(Exception exception) {
            onFatalError(exception);
        }
    }

    private final class ResourceConsumerLeaderListenerImpl implements ConsumerLeaderListener {

        @Override
        public void resourceConsumerGainedLeadership(
                final ConsumerID consumerId,
                final ResourceConsumerGateway consumerManagerGateway,
                final RCRPRegistrationSuccess registrationMessage) {
        	// 资源消费者领导权变更
            runAsync(
                    () ->
                            consumerTable.getConsumer(consumerId)
                                    .ifPresent(
                                            consumer ->
                                                    establishResourceConsumerConnection(
                                                            consumer,
                                                            consumerManagerGateway,
                                                            registrationMessage)));
        }

        @Override
        public void resourceConsumerLostLeadership(final ConsumerID consumerId, final ConsumingMasterId jobMasterId) {
            log.info(
                    "ConsumerManager for job {} with leader id {} lost leadership.", consumerId, jobMasterId);
            runAsync(
                    () ->
                            consumerTable.getConnection(consumerId)
                                    .ifPresent(
                                            jobManagerConnection ->
                                                    disconnectResourceConsumerConnection(
                                                            jobManagerConnection,
                                                            new Exception(
                                                                    "Job leader for job id "
                                                                            + consumerId
                                                                            + " lost leadership."))));
        }

        @Override
        public void handleError(Throwable throwable) {
            onFatalError(throwable);
        }

        @Override
        public void consumerManagerRejectedRegistration(
                ConsumerID jobId, String targetAddress, RCRPRegistrationRejection rejection) {
            runAsync(() -> handleRejectedJobManagerConnection(jobId, targetAddress, rejection));
        }
    }

    private final class ResourceManagerRegistrationListener
            implements RegistrationConnectionListener<
                    ResourceProviderToResourceManagerConnection,
                    ResourceProviderRegistrationSuccess,
                    ResourceProviderRegistrationRejection> {

        @Override
        public void onRegistrationSuccess(
                ResourceProviderToResourceManagerConnection connection,
                ResourceProviderRegistrationSuccess success) {
            final ResourceID resourceManagerId = success.getResourceManagerId();
            final InstanceID taskExecutorRegistrationId = success.getRegistrationId();
            final ClusterInformation clusterInformation = success.getClusterInformation();
            final ResourceManagerGateway resourceManagerGateway = connection.getTargetGateway();
            
            runAsync(
                    () -> {
                        // filter out outdated connections
                        //noinspection ObjectEquality
                        if (resourceManagerConnection == connection) {
                            try {
                                establishResourceManagerConnection(
                                        resourceManagerGateway,
                                        resourceManagerId,
                                        taskExecutorRegistrationId,
                                        clusterInformation);
                            } catch (Throwable t) {
                                log.error(
                                        "Establishing Resource Manager connection in ConsumeItem Executor failed",
                                        t);
                            }
                        }
                    });
        }

        @Override
        public void onRegistrationFailure(Throwable failure) {
            onFatalError(failure);
        }

        @Override
        public void onRegistrationRejection(
                String targetAddress, ResourceProviderRegistrationRejection rejection) {
            onFatalError(
                    new ERException(
                            String.format(
                                    "The TaskExecutor's registration at the ResourceManager %s has been rejected: %s",
                                    targetAddress, rejection)));
        }
    }

    private final class ProviderManagerActionsImpl implements ProviderManagerActions {
        private final ResourceConsumerGateway resourceConsumerGateway;

        private ProviderManagerActionsImpl(ResourceConsumerGateway jobMasterGateway) {
            this.resourceConsumerGateway = checkNotNull(jobMasterGateway);
        }

        @Override
        public void notifyFatalError(String message, Throwable cause) {
            try {
                log.error(message, cause);
            } catch (Throwable ignored) {
            }

            // The fatal error handler implementation should make sure that this call is
            // non-blocking
            fatalErrorHandler.onFatalError(cause);
        }

        @Override
        public void failTask(final ExecutionID executionId, final Throwable cause) {
            runAsync(() -> ResourceProvider.this.failTask(executionId, cause));
        }

        @Override
        public void updateTaskExecutionState(final ResourceAllocationState taskExecutionState) {

        }
    }

    private class SlotActionsImpl implements SlotActions {

        @Override
        public void freeSlot(final AllocationID allocationId) {
            runAsync(
                    () ->
                            freeSlotInternal(
                                    allocationId,
                                    new ERException(
                                            "TaskSlotTable requested freeing the TaskSlot "
                                                    + allocationId
                                                    + '.')));
        }

        @Override
        public void timeoutSlot(final AllocationID allocationId, final UUID ticket) {
            runAsync(() -> ResourceProvider.this.timeoutSlot(allocationId, ticket));
        }
    }

    private class ResourceConsumerHeartbeatListener
            implements HeartbeatListener<
                    AllocatedSlotReport, ResourceProviderToResourceConsumerHeartbeatPayload> {

        @Override
        public void notifyHeartbeatTimeout(final ResourceID resourceID) {
            final String message =
                    String.format(
                            "The heartbeat of JobManager with id %s timed out.",
                            resourceID.getStringWithMetadata());

            log.info(message);
            handleJobManagerConnectionLoss(resourceID, new TimeoutException(message));
        }

        @Override
        public void notifyTargetUnreachable(ResourceID resourceID) {
            final String message =
                    String.format(
                            "JobManager with id %s is no longer reachable.",
                            resourceID.getStringWithMetadata());
            log.info(message);

            handleJobManagerConnectionLoss(resourceID, new ResourceProviderException(message));
        }

        private void handleJobManagerConnectionLoss(ResourceID resourceID, Exception cause) {
            validateRunsInMainThread();
            consumerTable.getConnection(resourceID)
                    .ifPresent(
                            jobManagerConnection ->
                                    disconnectAndTryReconnectToResourceConsumer(
                                            jobManagerConnection, cause));
        }

        @Override
        public void reportPayload(ResourceID resourceID, AllocatedSlotReport allocatedSlotReport) {
            validateRunsInMainThread();
            OptionalConsumer.of(consumerTable.getConnection(allocatedSlotReport.getJobId()))
                    .ifPresent(
                            jobManagerConnection -> {
                                syncSlotsWithSnapshotFromJobMaster(
                                        jobManagerConnection.getResourceConsumerGateway(),
                                        allocatedSlotReport);
                            })
                    .ifNotPresent(
                            () ->
                                    log.debug(
                                            "Ignoring allocated slot report from job {} because there is no active leader.",
                                            allocatedSlotReport.getJobId()));
        }

        @Override
        public ResourceProviderToResourceConsumerHeartbeatPayload retrievePayload(ResourceID resourceID) {
            validateRunsInMainThread();
            return consumerTable.getConnection(resourceID)
                    .map(
                            jobManagerConnection -> {
                                ConsumerID jobId = jobManagerConnection.getConsumerId();

                                Set<ExecutionID> deployedExecutions = new HashSet<>();
                                List<AccumulatorSnapshot> accumulatorSnapshots = new ArrayList<>(16);
                                Iterator<ConsumingItem> allTasks = taskSlotTable.getTasks(jobId);

                                while (allTasks.hasNext()) {
                                    ConsumingItem consumingItem = allTasks.next();
                                    deployedExecutions.add(consumingItem.getExecutionId());

                                }
                                return new ResourceProviderToResourceConsumerHeartbeatPayload(
                                        new AccumulatorReport(accumulatorSnapshots),
                                        new ResourceConsumptionReport(deployedExecutions));
                            })
                    .orElseGet(ResourceProviderToResourceConsumerHeartbeatPayload::empty);
        }
    }

    private class ResourceManagerHeartbeatListener
            implements HeartbeatListener<Void, ResourceProviderHeartbeatPayload> {

        @Override
        public void notifyHeartbeatTimeout(final ResourceID resourceId) {
            final String message =
                    String.format(
                            "The heartbeat of ResourceManager with id %s timed out.",
                            resourceId.getStringWithMetadata());

            log.info(message);
            handleResourceManagerConnectionLoss(resourceId, new ResourceProviderException(message));
        }

        @Override
        public void notifyTargetUnreachable(ResourceID resourceID) {
            final String message =
                    String.format(
                            "ResourceManager with id %s is no longer reachable.",
                            resourceID.getStringWithMetadata());
            log.info(message);

            handleResourceManagerConnectionLoss(resourceID, new ResourceProviderException(message));
        }

        private void handleResourceManagerConnectionLoss(
                ResourceID resourceId, ResourceProviderException cause) {
            validateRunsInMainThread();

            // first check whether the timeout is still valid
            if (establishedResourceManagerConnection != null
                    && establishedResourceManagerConnection
                            .getResourceManagerResourceId()
                            .equals(resourceId)) {

                reconnectToResourceManager(cause);
            }
        }

        @Override
        public void reportPayload(ResourceID resourceID, Void payload) {
            // nothing to do since the payload is of type Void
        }

        @Override
        public ResourceProviderHeartbeatPayload retrievePayload(ResourceID resourceID) {
            validateRunsInMainThread();
            return new ResourceProviderHeartbeatPayload(
                    taskSlotTable.createSlotReport(getResourceID()));
        }
    }

    @VisibleForTesting
    static final class ResourceProviderConsumerServices implements ConsumerTable.ConsumerServices {

        private final Runnable closeHook;

        private ResourceProviderConsumerServices(Runnable closeHook) {
            this.closeHook = closeHook;
        }

        @Override
        public void close() {
            closeHook.run();
        }

        @VisibleForTesting
        static ResourceProviderConsumerServices create(Runnable closeHook) {
            return new ResourceProviderConsumerServices(closeHook);
        }
    }

	@Override
	public CompletableFuture<Acknowledge> submitCommand(CommandDescriptor cd, 
											ConsumingMasterId consumingMasterId, Time timeout) {
		// 检验AllocationId
		AllocationID allocationId = cd.getAllocationId();
		ConsumerID consumerId = cd.getConsumerId();
		ExecutionID executionId = cd.getExecutionId();
		
		try {
			// 获取消费者管理器连接，之前请求资源是注册
			final ConsumerTable.Connection consumerManagerConnection =
	                consumerTable.getConnection(consumerId)
	                        .orElseThrow(
	                                () -> {
	                                    final String message =
	                                            "Could not submit task because there is no ConsumerManager "
	                                                    + "associated for the consuming "
	                                                    + consumerId
	                                                    + '.';
	
	                                    log.debug(message);
	                                    return new ConsumingTaskSubmissionException(message);
	                                });
            // 验证当前JobMaster
            if (!Objects.equals(consumerManagerConnection.getConsumingMasterId(), consumingMasterId)) {
                final String message =
                        "Rejecting the command task submission because the consuming manager leader id "
                                + consumingMasterId
                                + " does not match the expected job manager leader id "
                                + consumerManagerConnection.getConsumingMasterId()
                                + '.';

                log.debug(message);
                throw new ConsumingTaskSubmissionException(message);
            }
            // 占用资源
            if (!taskSlotTable.tryMarkSlotActive(consumerId, allocationId)) {
                final String message =
                        "No task slot allocated for consumer ID "
                                + consumerId
                                + " and allocation ID "
                                + allocationId
                                + '.';
                log.debug(message);
                throw new ConsumingTaskSubmissionException(message);
            }
            //
            boolean taskAdded;
            try {
            	ConsumingItem item = new ConsumingItem(consumerId, executionId, allocationId);
                taskAdded = taskSlotTable.addConsumingTask(item);
            } catch (SlotNotFoundException | SlotNotActiveException e) {
                throw new ConsumingTaskSubmissionException("Could not submit command.", e);
            }
            // 
            if (taskAdded) {
    			String cmd = cd.getCommand();
    			String arg = cd.getSerArgs();
    			commandListener.execute(cmd, arg);
                return CompletableFuture.completedFuture(Acknowledge.get());
            } else {
                final String message =
                        "ProviderManager already contains a item for id " + cd.getExecutionId() + '.';
                log.debug(message);
                throw new ConsumingTaskSubmissionException(message);
            }
			
		} catch (ConsumingTaskSubmissionException e) {
            return FutureUtils.completedExceptionally(e);
        }

	}

}
